package com.atguigu.gmall.realtime.dws.app;

/*import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.codehaus.jackson.map.SerializationConfig;

import java.time.Duration;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(10023,  4, Constant.DWS_TRAFFIC_VC_CH_AR_ISNEW_PAGE_VIEW_WINDOW, Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 转换结构, 转成JSONOBJ
        SingleOutputStreamOperator<JSONObject> jsonObjDs = stream.map(JSON::parseObject);

        // jsonObjDs.print("input");


        // 处理 访问时长 页面浏览数 会话数 独立访客数
        // 基于每条数据,按照mid分组, 进行判断, 将结果封装成TrafficPageViewBean
        // 访问时长直接取
        // 页面浏览数 直接赋 1
        // 会话数 通过判断last_page_id是否为null, 如果是 表示当前是一次新的会话, 赋1 , 否则赋0
        // 独立访客数 通过状态来维护每个mid 的上次访问日期, 通过判断状态是否为null或者上次访问日期不为访日, 赋1, 否则, 赋0

        SingleOutputStreamOperator<TrafficPageViewBean> processDs = jsonObjDs.keyBy(
                // jsonObjDs -> jsonObjDs.getJSONObject("common").getString("mid")
                jsonObject -> jsonObject.getJSONObject("common").getString("mid")
        ).process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    // 定义状态, 维护mid上次访问日期
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastVisitDateState", Types.STRING);
                        // 独立访客每日重置，所以状态最多维护一天即可，为了减少内存开销，设置状态ttl为1天。
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                        // common
                        JSONObject commonObj = jsonObject.getJSONObject("common");

                        // page
                        JSONObject pageObj = jsonObject.getJSONObject("page");

                        // 上页id
                        String lastPageId = pageObj.getString("last_page_id");

                        // 从状态中取出上次访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        // 当前访问日期
                        Long ts = jsonObject.getLong("ts");
                        String currVisitDate = DateFormatUtil.tsToDate(ts);

                        // 访问时长
                        Long duringSum = pageObj.getLong("during_time");

                        // 页面浏览数
                        Long pvCt = 1L;

                        // 会话数
                        Long svCt = 0L;
                        if (lastPageId == null) {
                            svCt = 1L;
                        }

                        // 独立访客数
                        Long uvCt = 0L;
                        if (!currVisitDate.equals(lastVisitDate)) {
                            uvCt = 1L;
                            // 更新状态
                            lastVisitDateState.update(currVisitDate);
                        }

                        // 封装对象
                        TrafficPageViewBean trafficPageViewBean
                                = TrafficPageViewBean.builder()
                                .stt("")
                                .edt("")
                                .cur_date("")
                                .vc(commonObj.getString("vc"))
                                .ch(commonObj.getString("ch"))
                                .ar(commonObj.getString("ar"))
                                .isNew(commonObj.getString("is_new"))
                                .durSum(duringSum)
                                .pvCt(pvCt)
                                .svCt(svCt)
                                .uvCt(uvCt)
                                .ts(ts)
                                .build();

                        // 写出
                        collector.collect(trafficPageViewBean);

                    }
                }
        );
        // processDs.print();


        // 分配水位线
        SingleOutputStreamOperator<TrafficPageViewBean> timestampAndWatermarks = processDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                (bean, ts) -> bean.getTs()
                        )
        );


        // 通过版本,渠道,地区,访客类别 进行keyBy操作, 开窗, 聚合
        SingleOutputStreamOperator<TrafficPageViewBean> windowDs = timestampAndWatermarks.keyBy(
                bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + "_" + bean.getIsNew()
        ).window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))
        ).reduce(
                // 增量聚合
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean v1, TrafficPageViewBean v2) throws Exception {
                        v1.setDurSum(v1.getDurSum() + v2.getDurSum());
                        v1.setPvCt(v1.getPvCt() + v2.getPvCt());
                        v1.setSvCt(v1.getSvCt() + v2.getSvCt());
                        v1.setUvCt(v1.getUvCt() + v2.getUvCt());

                        return v1;
                    }
                },
                // 全窗口函数, 补充窗口信息
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>.Context context, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        // 提取窗口聚合的结果
                        // 因为先使用了增量聚合, 窗口的结果只有一条数据
                        TrafficPageViewBean trafficPageViewBean = iterable.iterator().next();

                        // 补充窗口信息
                        trafficPageViewBean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        trafficPageViewBean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(context.window().getStart()));

                        // 写出
                        collector.collect(trafficPageViewBean);
                    }
                }
        );
        windowDs.print();


        // 将结果写出到Doris表中   记得先在doris中建表
        windowDs
                // 就爱那个bean对象转换成Json格式
                .map(
                        // 并不能直接转, 要考虑将bean中的驼峰命名 转换成 doris表中的下划线命名
                        new BeanToJsonMapFunction()
                )
                .sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_TRAFFIC_VC_CH_AR_ISNEW_PAGE_VIEW_WINDOW)
                // .sinkTo(FlinkSinkUtil.getKafkaSink("haha123")
                );


    }
}*/
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author WEIYUNHUI
 * @date 2024/8/9 9:32
 */
public class DwsTrafficVcChArIsNewPageViewWindow  extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022 ,
                4 ,
                Constant.DWS_TRAFFIC_VC_CH_AR_ISNEW_PAGE_VIEW_WINDOW ,
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 转换结构 , 转成JSONObj
        SingleOutputStreamOperator<JSONObject> jsonObjDs = stream.map(JSON::parseObject);
        //jsonObjDs.print("INPUT");

        // 处理  访问时长  页面浏览数  会话数  独立访客数
        // 基于每条数据数据，按照mid分组，  进行判断处理， 将结果封装成 TrafficPageViewBean
        // 访问时长直接取
        // 页面浏览数直接赋 1
        // 会话数  通过判断 last_page_id 是否为null ，如果是，表示一次新的会话， 赋1 ， 否则赋0
        // 独立访客数  通过状态来维护每个mid的上次访问日期， 通过判断状态是否为null或者上次访问日期不为当日， 赋1 ， 否则，赋0

        SingleOutputStreamOperator<TrafficPageViewBean> processDs = jsonObjDs.keyBy(
                jsonObj -> jsonObj.getJSONObject("common").getString("mid")
        ).process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {

                    //定义状态， 维护mid上次访问日期
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateStateDesc =
                                new ValueStateDescriptor<>("lastVisitDateStateDesc", Types.STRING);
                        //TTL 状态存活时间为1天
                        lastVisitDateStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        //common
                        JSONObject commonObj = jsonObj.getJSONObject("common");

                        //page
                        JSONObject pageObj = jsonObj.getJSONObject("page");
                        //上页id
                        String lastPageId = pageObj.getString("last_page_id");

                        //从状态中取出上次访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        //当前访问日期
                        Long ts = jsonObj.getLong("ts");
                        String currVisitDate = DateFormatUtil.tsToDate(ts);

                        //访问时长
                        Long durSum = pageObj.getLong("during_time");

                        //页面浏览数
                        Long pvCt = 1L;

                        //会话数
                        Long svCt = 0L;
                        if (lastPageId == null) {
                            svCt = 1L;
                        }

                        //独立访客数
                        Long uvCt = 0L;
                        if (!currVisitDate.equals(lastVisitDate)) {
                            uvCt = 1L;
                            //更新状态
                            lastVisitDateState.update(currVisitDate);
                        }

                        //封装成TrafficPageViewBean对象
                        TrafficPageViewBean trafficPageViewBean
                                = TrafficPageViewBean.builder()
                                .stt("")
                                .edt("")
                                .cur_date("")
                                .vc(commonObj.getString("vc"))
                                .ch(commonObj.getString("ch"))
                                .ar(commonObj.getString("ar"))
                                .isNew(commonObj.getString("is_new"))
                                .durSum(durSum)
                                .pvCt(pvCt)
                                .svCt(svCt)
                                .uvCt(uvCt)
                                .ts(ts)
                                .build();

                        //写出
                        out.collect(trafficPageViewBean);

                    }
                }
        );

        //processDs.print() ;

        // 分配水位线
        SingleOutputStreamOperator<TrafficPageViewBean> timestampsAndWatermarks = processDs.assignTimestampsAndWatermarks(
                //WatermarkStrategy.forMonotonousTimestamps()
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                (bean, ts) -> bean.getTs()
                        )
        );

        // 通过 版本 渠道 地区 访客类别 进行keyBy, 开窗 、 聚合
        SingleOutputStreamOperator<TrafficPageViewBean> windowDs = timestampsAndWatermarks.keyBy(
                bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + "_" + bean.getIsNew()
        ).window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))
        ).reduce(
                //增量聚合
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());

                        return value1;
                    }
                },
                //全窗口函数， 补充窗口信息
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>.Context context, Iterable<TrafficPageViewBean> elements, Collector<TrafficPageViewBean> out) throws Exception {
                        // 提取窗口聚合的结果
                        // 因为先使用了增量聚合， 窗口的结果只有1条数据，
                        TrafficPageViewBean trafficPageViewBean = elements.iterator().next();

                        //补充窗口信息
                        trafficPageViewBean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        trafficPageViewBean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(context.window().getStart()));

                        //写出
                        out.collect(trafficPageViewBean);
                    }
                }
        );

        //windowDs.print();

        // 写出到Doris

        SingleOutputStreamOperator<String> map = windowDs
                // 将bean对象转换成json格式 ， 考虑将bean中的驼峰命名 处理成 表中的下划线命名
                .map(
                        new BeanToJsonMapFunction()
                );

        map.print("MAP");

        map.sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE , Constant.DWS_TRAFFIC_VC_CH_AR_ISNEW_PAGE_VIEW_WINDOW));

    }
}